package org.shj.spark.dataframe;

import org.apache.spark.api.java.function.ForeachFunction;
import org.apache.spark.sql.DataFrameReader;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * 实际企业级开发环境中，当传统数据库数据规模特别大时，如果采用传统的DB去处理的话，一般需要把数据分成很多批次处理，且实际的处理过程
 * 可能会非常复杂,通过传统的Java EE等技术可能很难或者不方便实现处理算法。此时采用SparkSQL获得数据库的数据并进行分布式处理就
 * 可以非常好的解决该问题。但是由于Spark SQL加载DB中的数据需要时间，所以一般会在Spark SQL和具体要操作的DB之间加上一个缓冲
 * 层次。 例如： 1. 中间使用Redis，处理速度可提高甚至45倍    2. 把数据库里数据导入到HDFS。 
 * 
 *
 */
public class SparkSQLMySql {

	public static void main(String[] args) {
		SparkSession ss = SparkSession.builder().appName("SparkSQLMySql").master("local").getOrCreate();
		
		DataFrameReader reader = ss.read().format("jdbc");
		reader.option("url", "jdbc:mysql://192.168.31.227:3306/jeeplus");
		reader.option("dbtable", "sys_user");
		reader.option("user", "root");
		reader.option("password", "123456");
		
		Dataset<Row> users = reader.load();
		users.createOrReplaceTempView("user");
		
		reader.option("dbtable", "sys_role");
		Dataset<Row> roles = reader.load();
		roles.createOrReplaceTempView("role");
		
		reader.option("dbtable", "sys_user_role");
		Dataset<Row> urmap = reader.load();
		urmap.createOrReplaceTempView("user_role");
		
		String sql = "select u.name, r.name from user_role ur, user u, role r "
				+ "where u.id=ur.user_id and r.id=ur.role_id";
		Dataset<Row> ds = ss.sql(sql);
		
		ds.foreach(new ForeachFunction<Row>() {
			private static final long serialVersionUID = 1L;

			public void call(Row t) throws Exception {
				System.out.println(t);
				
			}
		});
		
		
	}
}
